package pub.ryan.dw

import java.util.Properties

import ch.hsr.geohash.GeoHash
import org.apache.spark.sql.SparkSession

/**
 * 获取geoHash并存入本地文件夹data中
 */
object Dict_district {
  def main(args: Array[String]): Unit = {
    ///构造spark, 读数据库，获取lat,lng
    val spark = SparkSession.builder().appName(this.getClass.getSimpleName)
      .master(("local[*]"))
      .getOrCreate()
    import spark.implicits._

    val props = new Properties()
    props.setProperty("user", "root")
    props.setProperty("password", "root")

    val dataFrame = spark.read.jdbc("jdbc:mysql://c01:3307/dicts", "t_tmp", props)

    //将经纬度转为geohash字符串并输出到数据库中
    val result = dataFrame.map(row => {
      //取出这一行的经纬度
      val lat = row.getAs[Double]("BD09_LAT")
      val lng = row.getAs[Double]("BD09_LNG")
      val province = row.getAs[String]("province")
      val city = row.getAs[String]("city")
      val distinct = row.getAs[String]("distinct")

      //调用geohash算法，得到编码，精度越高，位置越细
      val geohash = GeoHash.geoHashStringWithCharacterPrecision(lat, lng, 5)
      //组装返回结果
      (province, city, distinct, geohash)
    }).toDF("province", "city", "distinct", "geohash")

    //写入本地data中
    result.write.parquet("data/dict/geo_dict/output")

    spark.close()
  }
}
